package com.yeming.flink.mudlog.function

import org.apache.flink.api.common.state.ValueState
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.util.Collector

/**
 * 定义一个函数类，用来保存用户上次接受数据的状态。
 */
class TimeDelayFunction extends ProcessFunction[(Long,Long), (Long,Long)] {
  // 定义一个状态用于保存上一次录井数据的记录
  private var preState: ValueState[(Long,Long)] = _

  override def processElement(value: (Long, Long), ctx: ProcessFunction[(Long, Long), (Long, Long)]#Context, out: Collector[(Long, Long)]): Unit = {
    if (value._1 == 0) {
      preState.update(value)
    } else {
      val pre: (Long, Long) = preState.value()
      println(s"初始值：${pre}")
      out.collect((value._1,(value._2-pre._2)/value._1))
    }
  }
}
